#include "store/RedisStore.h"
#include "utils/utils.h"

namespace zeroplus::mq
{

    std::shared_ptr<RedisStore> RedisStore::instance_ = nullptr;
    std::mutex RedisStore::mtx_;

    std::shared_ptr<RedisStore> RedisStore::getInstance(const std::string &uri)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if (!instance_)
        {
            instance_.reset(new RedisStore(uri));
        }
        return instance_;
    }

    RedisStore::RedisStore(const std::string &uri)
    {
        // 连接redis server
        int maxRetries = 50;
        int retries = 0;
        while (retries < maxRetries)
        {
            redis_.reset(new sw::redis::Redis(uri));
            try
            {
                auto res = redis_->ping();
                if (res == "PONG")
                {
                    LOG_INFO("Successfully connected to Redis: %s", uri.c_str());
                    break;
                }
            }
            catch (const std::exception &e)
            {
                LOG_WARN("Failed to connect to Redis: %s, retrying... (Attempt: %d)", uri.c_str(), ++retries);
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            }
        }

        if (retries == maxRetries)
        {
            LOG_ERROR("Failed to connect to Redis after %d attempts", maxRetries);
            exit(1);
        }
        // 实例化对象
        messageQueue_ = std::make_shared<MessageQueue>(redis_);
        delayQueue_ = std::make_shared<DelayQueue>(redis_);
        deadLetterQueue_ = std::make_shared<DeadLetterQueue>(redis_);
        CreateMQ("DLQ");
        auto handler = [this](const std::string &task) -> bool
        {
            DelayTaskInfo delatInfo = DelayTaskCodec::Decode(Bytes::from_string(task));
            WriteToMQ(delatInfo.topic, Bytes::from_string(delatInfo.message));
            return true;
        };
        delayQueue_->RegisterTaskCallback(handler);
        delayQueue_->Start();
    }

    RedisStore::~RedisStore()
    {
        delayQueue_->Stop();
    }

    std::string RedisStore::CreateMQ(const std::string &topic)
    {
        return messageQueue_->CreateStreams(topic);
    }

    std::string RedisStore::WriteToMQ(const std::string &topic, const Bytes &message)
    {
        return messageQueue_->WriteToStorage(topic, message);
    }

    std::tuple<Bytes, std::string, bool> RedisStore::ReadFromMQ(const std::string &topic, const std::string &consumerGroup, const std::string &consumerId,
                                                                const int &count, const int &blockSeconds)
    {
        std::shared_lock<std::shared_mutex> lock(rwMtx_);
        return messageQueue_->ReadFromStorage(topic, consumerGroup, consumerId, count, blockSeconds);
    }

    // 获取指定队列的大小
    long RedisStore::GetQueueSize(const std::string &topic)
    {
        return messageQueue_->Length(topic);
    }

    std::vector<std::string> RedisStore::GetAllTopic()
    {
        return messageQueue_->GetAllTopic();
    }

    // 检查key是否存在
    bool RedisStore::IsTopicExists(const std::string &topic)
    {
        return messageQueue_->IsTopicExists({topic});
    }

    bool RedisStore::WriteToDQ(const std::string &topic, const uint64_t &delay, const Bytes &message)
    {
        return delayQueue_->WriteToStorage(topic, delay, message);
    }

    bool RedisStore::AckMessage(const std::string &topic, const std::string &messageId, const std::string &groupName)
    {
        std::unique_lock<std::shared_mutex> lock(rwMtx_);
        return messageQueue_->Acknowledge(topic, messageId, groupName);
    }

    bool RedisStore::Reconsume(const std::string &topic, const std::string &consumerGroup, const std::string &consumerId, const std::string &messageId)
    {
        std::unique_lock<std::shared_mutex> lock(rwMtx_);
        std::vector<PendingItem> pendingItems = messageQueue_->SerachPendingMessage(topic, consumerGroup, consumerId, messageId);
        for (const auto &item : pendingItems)
        {
            std::string id;
            std::string consumerName;
            long long time;
            long long retryCount;

            std::tie(id, consumerName, time, retryCount) = item;
            if (retryCount < 5)
            {
                std::vector<ClaimedItem> claimedItem = messageQueue_->ClaimPendingMessage(topic, consumerName, consumerId, id);
                return true;
            }
            else
            {
                LOG_DEBUG("消息%s已进入死信队列, 请人工干预", id.c_str());
                auto [message, msgId, status] = messageQueue_->ReadFromStorage(topic, messageId);
                deadLetterQueue_->WriteToStorage(id, message);
                return false;
            }
        }
        LOG_DEBUG("no pending message!!!");
        return false;
    }
}
